/**
 * FileName: TestRedisAPI
 * Author:   SAMSUNG-PC 孙中军
 * Date:     2019/2/25 16:19
 * Description: Spark-redis 写入Redis并读取Redis
 */
package cn.com.bonc.app;

import cn.com.bonc.conf.ConfigurationManager;
import cn.com.bonc.constant.Constants;
import cn.com.bonc.domain.Xdata;
import cn.com.bonc.util.ExternalResourceUtil;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.io.*;
import java.util.List;

public class UploadData2Redis {

    private static List<Xdata> xdataList;

    public static void main(String[] args) throws FileNotFoundException {

        SparkSession spark = SparkSession
                .builder()
                .appName("MyRedisTestApp")
                .config("spark.redis.host", ConfigurationManager.getProperty(Constants.REDIS_IP))
                .config("spark.redis.port", ConfigurationManager.getProperty(Constants.REDIS_PORT))
                .getOrCreate();

        try {
            if (args.length>0){
                xdataList = ExternalResourceUtil.loadRedisTxtFileData(args[0]);
            }else {
                xdataList = ExternalResourceUtil.loadRedisTxtFileData();
            }
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("===============================>加载Redis.txt出错");
        }
        //以覆盖方式写入Redis
        spark.createDataFrame(xdataList, Xdata.class)
                .write()
                .format("org.apache.spark.sql.redis")
                .option("table", "Xdata")
                .option("key.column", "key")
                .mode(SaveMode.Overwrite)
                .save();

        //读取刚刚写入Redis的数据
        Dataset<Row> loadDf = spark.read()
                .format("org.apache.spark.sql.redis")
                .option("table", "Xdata")
                .load();

        //打印结果
        System.out.println("===============================>total:"+xdataList.size());
        System.out.println("=================================>star print redis schema");
        loadDf.printSchema();
        System.out.println("=================================>star print redis data");
        loadDf.show();
    }
}
